package com.bw.demo;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;

//实现Interceptor
public class ETLInterceptor implements Interceptor {
    //初始化拦截器，在拦截器启动时调用一次，用于资源分配
    @Override
    public void initialize() {

    }
    //处理单个事件 返回event对象 或 null丢弃该事件
    @Override
    public Event intercept(Event event) {
        //字节形式
        byte[] body = event.getBody();
        //                                   编码格式
        String bodyString = new String(body, StandardCharsets.UTF_8);
        if(JSONUtils.isJSON(bodyString)){
            return event; //是保留事件
        }else {
            return null;
        }
    }
    //批量处理事件列表,默认逐个调用intercept(Event)方法
    @Override
    public List<Event> intercept(List<Event> list) {
        //迭代器
        Iterator<Event> iterator = list.iterator();
        //迭代下一个数值
        while (iterator.hasNext()){
            Event next = iterator.next();
            //为空删除
            if(intercept(next)==null){
                iterator.remove();
            }
        }
       return list;
    }
    //关闭拦截器
    @Override
    public void close() {

    }
    //实现Interceptor.Builder接口
    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
